package com.niit.hadoop.maxandmin;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MaxAndMinReducer extends Reducer<MaxAndMinBean, NullWritable, MaxAndMinBean, NullWritable> {
    private long min = Long.MAX_VALUE;
    private long max = Long.MIN_VALUE;
    private int total = 0;

    @Override
    protected void reduce(MaxAndMinBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        if (key.getMin() < min) {
            min = key.getMin();
        }

        if (key.getMax() > max) {
            max = key.getMax();
        }

        total += key.getTotal();
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        MaxAndMinBean keyOut = new MaxAndMinBean(min, max, total);

        context.write(keyOut, NullWritable.get());
    }
}
